<?php
/**
 * Version: 1.0
 *
 * @category    SGN
 * @package     Lib
 * @subpackage	Queue
 * @copyright   Copyright (c) 2010 VNG (http://www.vng.com.vn)
 * @license     http://www.vng.com.vn
 * @version     $Id: Queue.php 1413 2012-02-29 07:10:26Z locth $
 * @author      $Author: locth $
 *
 */
require_once dirname(dirname(__FILE__)) .'/stompcli/Stomp.php';
class SGN_Queue
{
    /**
     *
     * Array of Stomp Client Objects
     *
     * @var array
     */
    private $_queue = array();
    /**
     * Connection Status Flag
     *
     *
     * @var boolean
     *
     */
    var $isConnected = false;
    /**
     *
     * Queue Config Object
     *
     * @var Zend Config
     */
    private $_config = null;
    /**
     *
     * @param $queueName $queueInstance
     */
    public function __construct ($queueName)
    {
        $this->_config = SGN_Application::getConfig('sgn', 'queue', $queueName)->toArray();
        $this->_queueName = $this->_config['queueName'];
        $this->_queue[$this->_queueName] = new Stomp($this->_config['host']);
    }
    /**
     *
     * @param string $queueName
     * @return vPortal_Queue
     */
    public static function factory ($queueName)
    {
        #$queue->connect($queueConfig['username'], $queueConfig['password']);
        return new SGN_Queue($queueName);
    }
    /**
     *
     * Connection to AMQ
     *
     *
     * @return void
     *
     */
    public function connect ()
    {
        if ($this->_isConnected[$this->_queueName]) {
            return;
        }
        $this->_isConnected[$this->_queueName] = true;
        try {
            $this->_queue[$this->_queueName]->connect(
            $this->_config['username'], $this->_config['password']);
        } catch (Exception $e) {
            $this->_isConnected[$this->_queueName] = false;
        }
    }
    /**
     *
     * Subscribe to a queue
     *
     * @return void
     *
     */
    public function subscribe ()
    {
        if ($this->_isSubscribed[$this->_queueName]) {
            return;
        }
        $this->_isSubscribed[$this->_queueName] = true;
        $this->_queue[$this->_queueName]->subscribe(
        "/queue/" . $this->_queueName);
    }
    /**
     *
     * Send a message to queue
     *
     * @param string $message
     *
     * @return void
     *
     */
    public function send ($message)
    {
        $this->connect();
        $this->_queue[$this->_queueName]->send(
        '/queue/' . $this->_config['queueName'], serialize($message),
        $this->_config['messageHeader']);
    }
    /**
     *
     * Reseive a message from queue
     *
     * @return string $msg
     *
     */
    public function receive ()
    {
        $this->connect();
        $this->subscribe();
        $msg = $this->_queue[$this->_queueName]->readFrame(true);
        if ($msg === false) { // stop read when no message in queue
            return;
        }
        if (! empty($msg)) {
            $message = unserialize($msg->body);
        }
        $this->_queue[$this->_queueName]->ack($msg);
        return $message;
    }
}